package com.niodata.dp.core.hdfs.legacy;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.DirectoryStream.Filter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * file system over hdfs.
 *
 * @author huqiao
 */
public class HdfsFileSystem implements DpFileSystem {

  private final RawHdfsFileSystem rawHdfs;
  private final Logger logger = LoggerFactory.getLogger(getClass());
  private final String realuser;

  private HdfsFileSystem(String realuser) throws IOException {
    this.rawHdfs = RawHdfsFileSystem.getInstance(realuser);
    this.realuser = realuser;
  }

  public static HdfsFileSystem getInstance(String realuser) throws IOException {
    HdfsFileSystem fs = new HdfsFileSystem(realuser);
    return fs;
  }

  @Override
  public List<DpFileStatus> listFiles(String path, Integer limit, Integer page)
        throws IOException, DpFsException {
    return listFiles(path, null, limit, page);
  }

  /**
   * list files.
   */
  @Override
  public List<DpFileStatus> listFiles(String path, Filter<DpFileStatus> filter,
        Integer limit, Integer page) throws IOException {

    List<DpFileStatus> files = new ArrayList<DpFileStatus>();
    FileStatus[] hdfsFiles = rawHdfs.listStatus(resetPath(path));
    if (hdfsFiles != null && hdfsFiles.length > 0) {
      for (FileStatus f : hdfsFiles) {
        DpFileStatus gfs = createDpFileStatus(f);
        if (filter == null || filter.accept(gfs)) {
          files.add(gfs);
        }
      }
    }

    int fromIndex = 0;
    int toIndex = files.size();
    if (limit != null && limit > 0) {
      fromIndex = limit * ((page == null || page < 0) ? 0 : page);
      toIndex = fromIndex + limit;
    }
    fromIndex = fromIndex > files.size() ? files.size() : fromIndex;
    toIndex = toIndex > files.size() ? files.size() : toIndex;

    return files.subList(fromIndex, toIndex);
  }


  private DpFilePermission getDpFilePermission(FileStatus file) {

    FsPermission permission = file.getPermission();

    DpFileAction groupAction = createDpFileAction(permission.getGroupAction());

    DpFileAction otherAction = createDpFileAction(permission.getOtherAction());

    DpFilePermission dpPermission = new DpFilePermission(groupAction, otherAction);

    return dpPermission;
  }

  private DpFileAction createDpFileAction(FsAction fsAction) {

    DpFileAction gfAction = DpFileAction.NONE;

    if (fsAction.implies(FsAction.READ) && fsAction.implies(FsAction.WRITE)) {
      gfAction = DpFileAction.WRITE_READ;
    } else if (fsAction.implies(FsAction.WRITE)) {
      gfAction = DpFileAction.WRITE;
    } else if (fsAction.implies(FsAction.READ)) {
      gfAction = DpFileAction.READ;
    }

    return gfAction;
  }

  private DpFileStatus createDpFileStatus(FileStatus f) {
    String path = Path.getPathWithoutSchemeAndAuthority(f.getPath()).toString();
    DpFilePermission perm = getDpFilePermission(f);
    Date createTime = new Date(f.getModificationTime());
    DpFileStatus gfs = new DpFileStatus(path, perm, f.getLen(), f.getOwner(), f.getGroup(),
          createTime, f.isDirectory());
    return gfs;
  }

  @Override
  public void mkdir(String path) throws IOException {
    rawHdfs.mkdir(resetPath(path), createFsPermission(DpFilePermission.getDirDefault()));
  }

  private FsPermission createFsPermission(DpFilePermission dpPermission) {
    FsAction userAction = FsAction.ALL;
    FsAction groupAction = createFsAction(dpPermission.getGroupAction());
    FsAction otherAction = createFsAction(dpPermission.getOtherAction());
    FsPermission fsPerm = new FsPermission(userAction, groupAction, otherAction);
    return fsPerm;
  }

  private FsAction createFsAction(DpFileAction dpAction) {
    if (DpFileAction.WRITE_READ.equals(dpAction)) {
      return FsAction.READ_WRITE;
    } else if (DpFileAction.WRITE.equals(dpAction)) {
      return FsAction.WRITE;
    } else if (DpFileAction.READ.equals(dpAction)) {
      return FsAction.READ;
    }
    return FsAction.NONE;
  }

  @Override
  public void delete(String path, boolean recursive) throws IOException {
    rawHdfs.delete(resetPath(path), recursive);
  }

  @Override
  public void copyFromLocal(String src, String target) throws IOException {
    rawHdfs.copyFromLocal(src, resetPath(target),
          createFsPermission(DpFilePermission.getFileDefault()));
  }

  @Override
  public void copyToLocal(String src, String target) throws IOException {
    rawHdfs.copyToLocal(resetPath(src), target);
  }

  @Override
  public boolean isDirectory(String path) throws IOException {
    return rawHdfs.isDirectory(resetPath(path));
  }

  @Override
  public long getCapacity() throws IOException {
    return rawHdfs.getCapacity();
  }

  @Override
  public long getUsed() throws IOException {
    return rawHdfs.getUsed();
  }

  @Override
  public boolean exist(String path) throws IOException {
    return rawHdfs.exists(resetPath(path));
  }

  @Override
  public boolean isFile(String path) throws IOException {
    return rawHdfs.isFile(resetPath(path));
  }

  @Override
  public long getFileSize(String path) throws IOException {
    return rawHdfs.getFileSize(resetPath(path));
  }

  @Override
  public long write(String path, InputStream os, boolean overwrite) throws IOException {
    String realPath = resetPath(path);
    // make directory
    String dir = new Path(realPath).getParent().toString();
    if (!rawHdfs.exists(dir)) {
      logger.info("mkdir {}", dir);
      rawHdfs.mkdirs(dir, createFsPermission(DpFilePermission.getDirDefault()));
    }
    return rawHdfs.write(realPath, os, overwrite,
          createFsPermission(DpFilePermission.getFileDefault()));
  }

  @Override
  public InputStream open(String path) throws IOException {
    return rawHdfs.open(resetPath(path));
  }

  @Override
  public OutputStream create(String path, boolean overwrite) throws IOException {
    return rawHdfs.create(path, overwrite);
  }

  @Override
  public void close() throws IOException {
    rawHdfs.close();
  }

  @Override
  public void mkdirs(String path) throws IOException {
    rawHdfs.mkdirs(resetPath(path), createFsPermission(DpFilePermission.getDirDefault()));
  }

  private String resetPath(String orgPath) {
    return orgPath;
  }
}